/**
 * Copyright (C) 2012-2013 jumei, Inc.
 * TestClient.java
 * 2015年6月8日
 */
package org.goface.rpc.io;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPromise;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

import akka.remote.serialization.ProtobufSerializer;

import com.example.tutorial.AddressBookProtos.Person;

/**
 * @author jm110
 */
public class TestClient {
    private static BusMessage createMsg() {
        BusMessage msg = new BusMessage();
        msg.age = 10;
        msg.kl = 20;
        msg.money = 10.01f;
        msg.name = "test name";
        msg.phones = new LinkedList<String>();
        msg.phones.add("12585642");
        msg.phones.add("17895268");
        msg.ratio = 20.998;
        msg.relations = new HashMap<>();
        msg.relations.put("r1", msg);
        msg.relations.put("r2", null);
        msg.right = true;
        msg.sigle = 'c';
        msg.type = 9;
        return msg;

    }

    public static class BusMessage implements Serializable {
        private static final long serialVersionUID = 1L;
        private String name;
        private int age;
        private List<String> phones;
        private Map<String, BusMessage> relations;
        private float money;
        private double ratio;
        private byte type;
        private char sigle;
        private boolean right;
        private short kl;

        /**
         * @return the name
         */
        public String getName() {
            return name;
        }

        /**
         * @param name
         *            the name to set
         */
        public void setName(String name) {
            this.name = name;
        }

        /**
         * @return the age
         */
        public int getAge() {
            return age;
        }

        /**
         * @param age
         *            the age to set
         */
        public void setAge(int age) {
            this.age = age;
        }

        /**
         * @return the phones
         */
        public List<String> getPhones() {
            return phones;
        }

        /**
         * @param phones
         *            the phones to set
         */
        public void setPhones(List<String> phones) {
            this.phones = phones;
        }

        /**
         * @return the relations
         */
        public Map<String, BusMessage> getRelations() {
            return relations;
        }

        /**
         * @param relations
         *            the relations to set
         */
        public void setRelations(Map<String, BusMessage> relations) {
            this.relations = relations;
        }

        /**
         * @return the money
         */
        public float getMoney() {
            return money;
        }

        /**
         * @param money
         *            the money to set
         */
        public void setMoney(float money) {
            this.money = money;
        }

        /**
         * @return the ratio
         */
        public double getRatio() {
            return ratio;
        }

        /**
         * @param ratio
         *            the ratio to set
         */
        public void setRatio(double ratio) {
            this.ratio = ratio;
        }

        /**
         * @return the type
         */
        public byte getType() {
            return type;
        }

        /**
         * @param type
         *            the type to set
         */
        public void setType(byte type) {
            this.type = type;
        }

        /**
         * @return the sigle
         */
        public char getSigle() {
            return sigle;
        }

        /**
         * @param sigle
         *            the sigle to set
         */
        public void setSigle(char sigle) {
            this.sigle = sigle;
        }

        /**
         * @return the right
         */
        public boolean isRight() {
            return right;
        }

        /**
         * @param right
         *            the right to set
         */
        public void setRight(boolean right) {
            this.right = right;
        }

        /**
         * @return the kl
         */
        public short getKl() {
            return kl;
        }

        /**
         * @param kl
         *            the kl to set
         */
        public void setKl(short kl) {
            this.kl = kl;
        }

    }

    private static Person createPerson() {
        Person john = Person.newBuilder().setId(1234).setName("John Doe").setEmail("jdoe@example.com")
                .addPhone(Person.PhoneNumber.newBuilder().setNumber("555-4321").setType(Person.PhoneType.HOME)).build();
        return john;
    }

    static AtomicLong succ = new AtomicLong(0);
    static CountDownLatch counter = new CountDownLatch(1);
    static int number = 1000000;

    public static void main1(String[] args) throws InterruptedException, ExecutionException {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup(2, Executors.newFixedThreadPool(2)));
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                NettyUtils.addJavaSerialiseCodec(ch, new MyHandler());
            }

        });
        ChannelFuture future = bootstrap.connect(new InetSocketAddress("127.0.0.1", 5125));
        future.sync();
        System.out.println("connected");
        Channel channel = future.channel();
        // BusMessage msg = createMsg();
        GenericFutureListener gl = new GenericFutureListener<Future<? super Void>>() {

            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.cause() != null) {
                    future.cause().printStackTrace();
                }
            }
        };
        long t1 = System.currentTimeMillis();
        for (int i = 0; i < number; ++i) {
            Person person = createPerson();
            ChannelFuture res = channel.writeAndFlush(person);
            res.addListener(gl);
        }
        counter.await();
        long t2 = System.currentTimeMillis();
        System.out.println("use time: " + (t2 - t1) / 1000);
        // res.cause().printStackTrace();
    }

    static class MyHandler extends ChannelHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            Person busMsg = (Person) msg;
            succ.incrementAndGet();
            if (succ.get() >= number) {
                counter.countDown();
            }
            // System.out.println(busMsg.getId() + ": " + busMsg.getEmail());
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            // Close the connection when an exception is raised.
            cause.printStackTrace();
            ctx.close();
        }

        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            // System.out.println("write!!!!!!");
            ctx.writeAndFlush(msg, promise);
        }
    }

    public static void main(String[] args) {
        ProtobufSerializer serializer = new ProtobufSerializer();
        BusMessage msg = createMsg();
        byte[] objectbytes = serializer.toBinary(msg);
        Object messageObj = serializer.fromBinary(objectbytes);
        System.out.println("messageObj"+messageObj);

    }
}
